package org.jeecg.modules.iot.aio.task;

import com.alibaba.fastjson.JSON;
import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import org.jeecg.modules.iot.aio.AioConfiguration;
import org.jeecg.modules.iot.aio.message.CheckMessage;
import org.jeecg.modules.iot.aio.task.base.AbstractQueueRunnable;
import org.jeecg.modules.iot.aliyun.MessageBody;
import org.jeecg.modules.iot.tools.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.TcpConst;
import org.tio.utils.queue.FullWaitQueue;
import org.tio.utils.queue.TioFullWaitQueue;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * 这里是一个注释
 *
 * @author scw
 * @version v1.0
 * @date
 */
public class QueryMessageRunnable extends AbstractQueueRunnable<CheckMessage> {

    private static final Logger log	 = LoggerFactory.getLogger(QueryMessageRunnable.class);

    /** The msg queue. */
    private FullWaitQueue<CheckMessage> msgQueue = null;
    public boolean canSend = true;

    public AioConfiguration aioConfiguration;

    private static final int	MAX_CAPACITY_MIN	= TcpConst.MAX_DATA_LENGTH - 1024;	//减掉1024是尽量防止溢出的一小部分还分成一个tcp包发出
    private static final int	MAX_CAPACITY_MAX	= MAX_CAPACITY_MIN * 10;

    public QueryMessageRunnable(ThreadPoolExecutor executor, AioConfiguration aioConfiguration) {
        super(executor);
        this.aioConfiguration = aioConfiguration;
    }

    @Override
    public void runTask() {
//        if (getMsgQueue().isEmpty())return;
        String endPoint = "http://" + aioConfiguration.getIotClientUtil().getMnsEndpoint();
        CloudAccount account = new CloudAccount(aioConfiguration.getIotClientUtil().getAccessKeyId(), aioConfiguration.getIotClientUtil().getAccessKeySecret(), endPoint);
        MNSClient client = account.getMNSClient();
        CloudQueue queue = client.getQueueRef("aliyun-iot-" + aioConfiguration.getProductKey()); //参数请输入IoT自动创建的队列名称
        Message popMsg = null;
        while (!Thread.interrupted()) {
            try {
                // 获取消息
                popMsg = queue.popMessage(10); //长轮询等待时间为10秒
                if (popMsg != null) {
                    MessageBody messageBody = JsonUtils.toObject(popMsg.getMessageBody(), MessageBody.class);
                    aioConfiguration.handlerMessage(messageBody);
                }
            }
            catch (Exception e) {
                e.printStackTrace();
                aioConfiguration.setException(e);
                if(e.getCause() instanceof InterruptedException){
                    break;
                }
            } finally {
                if (queue != null && popMsg != null) {
                    try {
                        //从队列中删除消息
                        queue.deleteMessage(popMsg.getReceiptHandle());
                        log.info("删除消息："+ JSON.toJSONString(popMsg.getMessageBody()));
                    } catch (Exception e) {
                        e.printStackTrace();
                        aioConfiguration.setException(e);
                    } finally {
                        popMsg = null;
                    }
                }
            }
        }
    }


    @Override
    public FullWaitQueue<CheckMessage> getMsgQueue() {
        if (msgQueue == null) {
            synchronized (this) {
                if (msgQueue == null) {
                    msgQueue = new TioFullWaitQueue<CheckMessage>(Integer.getInteger("tio.fullqueue.capacity", null), false);
                }
            }
        }
        return msgQueue;
    }

}
